package org.xiaoxin.onecoupon.engine.mq.consumer;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.xiaoxin.onecoupon.engine.common.constant.EngineRedisConstant;
import org.xiaoxin.onecoupon.engine.common.context.UserContext;
import org.xiaoxin.onecoupon.engine.mq.event.CanalBinlogEvent;
import org.xiaoxin.onecoupon.engine.mq.event.UserCouponDelayCloseEvent;
import org.xiaoxin.onecoupon.engine.mq.producer.UserCouponDelayCloseProducer;

import java.lang.ref.PhantomReference;
import java.util.Map;

/**
 * @author xiaoxin
 * 2025/5/13
 */
@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = "one-coupon_canal_engine-service_common-sync_topic${unique-name:}",
        consumerGroup = "one-coupon_canal_engine-service_common-sync_cg${unique-name:}"
)
public class CanalBinlogSyncUserCouponConsumer implements RocketMQListener<CanalBinlogEvent> {
   private final StringRedisTemplate stringRedisTemplate;
   private final UserCouponDelayCloseProducer userCouponDelayCloseProducer;
   @Override
   public void onMessage(CanalBinlogEvent canalBinlogEvent) {
      log.info("canal正在消费:");
      Map<String, Object> first = CollUtil.getFirst(canalBinlogEvent.getData());
      String couponTemplateId = first.get("coupon_template_id").toString();
      String userCouponId = first.get("id").toString();
      //用户创建优惠劵事件
      if (ObjectUtil.equal(canalBinlogEvent.getType(),"INSERT")){
         //将用户领取优惠劵记录添加到redis缓存中
         String userCouponListCacheKey = String.format(EngineRedisConstant.USER_COUPON_TEMPLATE_LIST_KEY, UserContext.getUserId());
         String userCouponItemCacheKey = StrUtil.builder()
                 .append(couponTemplateId)
                 .append("_")
                 .append(userCouponId)
                 .toString();
         DateTime receiveTime = DateUtil.parse(first.get("receive_time").toString());
         stringRedisTemplate.opsForZSet().add(userCouponListCacheKey,userCouponItemCacheKey, receiveTime.getTime());
         // 由于 Redis 在持久化或主从复制的极端情况下可能会出现数据丢失，而我们对指令丢失几乎无法容忍，因此我们采用经典的写后查询策略来应对这一问题
         Double scored;
         try {
            scored=stringRedisTemplate.opsForZSet().score(userCouponListCacheKey,userCouponItemCacheKey);
            if (scored==null){
               stringRedisTemplate.opsForZSet().add(userCouponListCacheKey,userCouponItemCacheKey,receiveTime.getTime());
            }
         }catch (Throwable ex){
            log.warn("查询Redis用户优惠券记录为空或抛异常，可能Redis宕机或主从复制数据丢失，基础错误信息：{}", ex.getMessage());
            // 如果直接抛异常大概率 Redis 宕机了，所以应该写个延时队列向 Redis 重试放入值。为了避免代码复杂性，这里直接写新增，知道最优解决方案即可
            stringRedisTemplate.opsForZSet().add(userCouponListCacheKey, userCouponItemCacheKey, receiveTime.getTime());
         }
         // 发送延时消息队列，等待优惠券到期后，将优惠券信息从缓存中删除
         UserCouponDelayCloseEvent userCouponDelayCloseEvent = UserCouponDelayCloseEvent.builder()
                 .couponTemplateId(couponTemplateId)
                 .userId(UserContext.getUserId())
                 .userCouponId(userCouponId)
                 .delayTime(DateUtil.parse(first.get("valid_end_time").toString()).getTime())
                 .build();
         SendResult sendResult = userCouponDelayCloseProducer.sendMessage(userCouponDelayCloseEvent);
         // 发送消息失败解决方案简单且高效的逻辑之一：打印日志并报警，通过日志搜集并重新投递
         if (ObjectUtil.notEqual(sendResult.getSendStatus().name(), "SEND_OK")) {
            log.warn("发送优惠券关闭延时队列失败，消息参数：{}", JSON.toJSONString(userCouponDelayCloseEvent));
         }
      }
   }
}
